package com.zc.finkDemo.source;

import com.zc.finkDemo.entities.ErrorMessage;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.UUID;

public class KafkaSource extends RichParallelSourceFunction<ErrorMessage> {

    private Boolean sendMessage = true;

    private Random random = new Random();

    private List<String> ecsList = Arrays.asList("A", "B", "C", "D", "E");

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("source建立kafka连接");

    }

    @Override
    public void close() throws Exception {
        System.out.println("source断开kafka连接");
    }

    @Override
    public void run(SourceContext<ErrorMessage> sourceContext) throws Exception {
        //模拟消息的产生
        while (sendMessage) {
            Thread.sleep(700);
            long l = System.currentTimeMillis();
            ErrorMessage errorMessage = new ErrorMessage();
            errorMessage.setMessageId(l);
            errorMessage.setType(random.nextInt(2));
            errorMessage.setErrorDevice(UUID.randomUUID().toString().substring(30));
            errorMessage.setParentECS(ecsList.get(random.nextInt(4)));
            errorMessage.setIp("127.0.0.1");
            errorMessage.setErrorReason("网络异常");
            errorMessage.setCreateTime(l);
            sourceContext.collect(errorMessage);
        }
    }

    @Override
    public void cancel() {
        this.sendMessage = false;
    }
}
